package cn.zl.springbootkafka.consumer;

import cn.zl.springbootkafka.product.KafkaProducter;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @Description:Kafka消费者
 * @Author: zllwsy
 * @Date: 2020/10/28 16:30
 */
@Component
@Slf4j
public class KafkaConsumer {
    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);

    @KafkaListener(topics = KafkaProducter.TOPIC_ONE,groupId = KafkaProducter.TOPIC_Group1)
    public void topic_test1(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic){
        Optional message=Optional.ofNullable(record.value());
        if (message.isPresent()){
            Object msg=message.get();
            logger.info("topic_test1 消费了： +++++++++++++++ Topic:" + topic+",Record:" + record+",Message:" + msg);
            ack.acknowledge();
        }
    }

    @KafkaListener(topics = KafkaProducter.TOPIC_ONE,groupId = KafkaProducter.TOPIC_Group2)
    public void topic_test2(ConsumerRecord<?, ?> record, Acknowledgment ack, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic){
        Optional message=Optional.ofNullable(record.value());
        if (message.isPresent()){
            Object msg=message.get();
            logger.info("topic_test2 消费了： +++++++++++++++ Topic:" + topic+",Record:" + record+",Message:" + msg);
            ack.acknowledge();
        }
    }
}
